/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.cassandra.service.accord;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;

import org.junit.Assert;
import org.junit.Test;

import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
import org.apache.cassandra.utils.concurrent.AsyncFuture;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;

public class AccordAsyncConversionTest
{
    private static final String SUCCESS = "test-success";
    private static final RuntimeException EXCEPTION = new RuntimeException("test-failure");

    @Test
    public void testAsyncChainToFutureSuccess()
    {
        AsyncChain<String> chain = AsyncChains.success(SUCCESS);
        Future<String> future = AccordService.toFuture(chain);
        
        Assert.assertTrue("should be completed", future.isDone());
        Assert.assertTrue("should be successful", future.isSuccess());
        Assert.assertEquals("should contain expected value", SUCCESS, future.getNow());
    }

    @Test
    public void testAsyncChainToFutureFailure()
    {
        AsyncChain<String> chain = AsyncChains.failure(EXCEPTION);
        Future<String> future = AccordService.toFuture(chain);
        
        Assert.assertTrue("should be completed", future.isDone());
        Assert.assertFalse("should be failed", future.isSuccess());
        Assert.assertEquals("should contain expected exception", EXCEPTION, future.cause());
    }

    @Test
    public void testAsyncChainToFutureAsyncCompletion() throws InterruptedException
    {
        CountDownLatch startLatch = new CountDownLatch(1);
        AsyncChain<String> chain = new AsyncChains.Head<String>()
        {
            @Override
            protected accord.utils.async.Cancellable start(BiConsumer<? super String, Throwable> callback)
            {
                Thread t = new Thread(() -> {
                    try
                    {
                        startLatch.await();
                        callback.accept(SUCCESS, null);
                    }
                    catch (InterruptedException e)
                    {
                        callback.accept(null, e);
                    }
                });
                t.start();
                return () -> t.interrupt();
            }
        };
        
        Future<String> future = AccordService.toFuture(chain);
        
        Assert.assertFalse("should not be completed initially", future.isDone());
        
        startLatch.countDown();
        
        Assert.assertEquals("should complete with expected value", SUCCESS, future.syncUninterruptibly().getNow());
    }

    @Test
    public void testAsyncResultToFutureWhenAlreadyFuture()
    {
        AsyncFuture<String> original = ImmediateFuture.success(SUCCESS);
        Future<String> converted = AccordService.toFuture(original);
        
        Assert.assertSame("Should return the same instance when already a Future", original, converted);
    }

    @Test
    public void testAsyncResultToFutureSuccess()
    {
        AsyncResult<String> result = new TestAsyncResult<>(SUCCESS, null);
        Future<String> future = AccordService.toFuture(result);
        
        Assert.assertTrue("should be completed", future.isDone());
        Assert.assertTrue("should be successful", future.isSuccess());
        Assert.assertEquals("should contain expected value", SUCCESS, future.getNow());
    }

    @Test
    public void testAsyncResultToFutureFailure()
    {
        AsyncResult<String> result = new TestAsyncResult<>(null, EXCEPTION);
        Future<String> future = AccordService.toFuture(result);
        
        Assert.assertTrue("should be completed", future.isDone());
        Assert.assertFalse("should be failed", future.isSuccess());
        Assert.assertEquals("should contain expected exception", EXCEPTION, future.cause());
    }

    @Test
    public void testGetBlockingAsyncChainSuccess()
    {
        AsyncChain<String> chain = AsyncChains.success(SUCCESS);
        String result = AccordService.getBlocking(chain);
        
        Assert.assertEquals("Should return expected value", SUCCESS, result);
    }

    @Test(expected = RuntimeException.class)
    public void testGetBlockingAsyncChainFailure()
    {
        AsyncChain<String> chain = AsyncChains.failure(EXCEPTION);
        AccordService.getBlocking(chain);
    }

    @Test
    public void testGetBlockingAsyncResultSuccess()
    {
        AsyncResult<String> result = new TestAsyncResult<>(SUCCESS, null);
        String value = AccordService.getBlocking(result);
        
        Assert.assertEquals("Should return expected value", SUCCESS, value);
    }

    @Test(expected = RuntimeException.class)
    public void testGetBlockingAsyncResultFailure()
    {
        AsyncResult<String> result = new TestAsyncResult<>(null, EXCEPTION);
        AccordService.getBlocking(result);
    }

    @Test
    public void testAsyncChainCancellation() throws InterruptedException
    {
        AtomicBoolean cancelled = new AtomicBoolean(false);
        CountDownLatch startLatch = new CountDownLatch(1);
        
        AsyncChain<String> chain = new AsyncChains.Head<String>()
        {
            @Override
            protected accord.utils.async.Cancellable start(BiConsumer<? super String, Throwable> callback)
            {
                Thread t = new Thread(() -> {
                    try
                    {
                        startLatch.await();
                        Thread.sleep(1000);
                        callback.accept(SUCCESS, null);
                    }
                    catch (InterruptedException e)
                    {
                        cancelled.set(true);
                        callback.accept(null, e);
                    }
                });
                t.start();
                return () -> {
                    cancelled.set(true);
                    t.interrupt();
                };
            }
        };
        
        Future<String> future = AccordService.toFuture(chain);
        startLatch.countDown();
        
        boolean cancelResult = future.cancel(true);
        Thread.sleep(100);
        
        Assert.assertTrue("should be cancellable", cancelResult);
        Assert.assertTrue("should be cancelled", future.isCancelled());
    }

    @Test
    public void testMultipleCallbacksOnAsyncResult() throws InterruptedException
    {
        AtomicReference<String> callback1Result = new AtomicReference<>();
        AtomicReference<String> callback2Result = new AtomicReference<>();
        CountDownLatch callbackLatch = new CountDownLatch(2);
        
        AsyncResult<String> result = new TestAsyncResult<>(SUCCESS, null);
        
        result.invoke((value, throwable) -> {
            callback1Result.set(value);
            callbackLatch.countDown();
        });
        
        result.invoke((value, throwable) -> {
            callback2Result.set(value);
            callbackLatch.countDown();
        });
        
        Assert.assertTrue("Both callbacks should be invoked", callbackLatch.await(1, TimeUnit.SECONDS));
        Assert.assertEquals("First callback should receive correct value", SUCCESS, callback1Result.get());
        Assert.assertEquals("Second callback should receive correct value", SUCCESS, callback2Result.get());
    }

    @Test
    public void testNullAsyncResult()
    {
        try
        {
            AccordService.toFuture((AsyncResult<String>) null);
            Assert.fail("Should throw NullPointerException for null AsyncResult");
        }
        catch (NullPointerException e)
        {
        }
    }

    @Test
    public void testNullAsyncChain()
    {
        try
        {
            AccordService.toFuture((AsyncChain<String>) null);
            Assert.fail("Should throw NullPointerException for null AsyncChain");
        }
        catch (NullPointerException e)
        {
            // ignore
        }
    }

    @Test
    public void testFutureInterfaceCompliance()
    {
        AsyncChain<String> chain = AsyncChains.success(SUCCESS);
        Future<String> future = AccordService.toFuture(chain);
        
        Assert.assertTrue("should implement AsyncResult", future instanceof AsyncResult);
        Assert.assertTrue("should be completion-aware", future.isDone());
        Assert.assertNotNull("should have a result", future.getNow());
    }

    @Test
    public void testConcurrentConversion() throws InterruptedException
    {
        int threadCount = 10;
        CountDownLatch startLatch = new CountDownLatch(1);
        CountDownLatch completeLatch = new CountDownLatch(threadCount);
        AtomicReference<Exception> exception = new AtomicReference<>();
        
        for (int i = 0; i < threadCount; i++)
        {
            final int threadIndex = i;
            Thread t = new Thread(() -> {
                try
                {
                    startLatch.await();
                    AsyncChain<Integer> chain = AsyncChains.success(threadIndex);
                    Future<Integer> future = AccordService.toFuture(chain);
                    Assert.assertEquals("Thread " + threadIndex + " should get correct result",
                                      Integer.valueOf(threadIndex), future.getNow());
                }
                catch (Exception e)
                {
                    exception.set(e);
                }
                finally
                {
                    completeLatch.countDown();
                }
            });
            t.start();
        }
        
        startLatch.countDown();
        Assert.assertTrue("All threads should complete", completeLatch.await(5, TimeUnit.SECONDS));
        Assert.assertNull("No exceptions should occur during concurrent conversion", exception.get());
    }

    private static class TestAsyncResult<V> implements AsyncResult<V>
    {
        private final V successValue;
        private final Throwable failureValue;

        public TestAsyncResult(V successValue, Throwable failureValue)
        {
            this.successValue = successValue;
            this.failureValue = failureValue;
        }

        @Override
        public boolean isDone()
        {
            return true;
        }

        @Override
        public boolean isSuccess()
        {
            return false;
        }

        @Override
        public AsyncResult<V> invoke(BiConsumer<? super V, Throwable> callback)
        {
            callback.accept(successValue, failureValue);
            return null;
        }
    }
}